package com.xy.job.admin.core.schedule

import com.xy.job.admin.common.PRE_SCHEDULE_SECONDS
import com.xy.job.admin.core.InetAddressHolder
import com.xy.job.admin.domain.JobInfo
import com.xy.job.admin.domain.JobInstance
import com.xy.job.admin.enums.ScheduleType
import com.xy.job.admin.enums.ScheduleType.*
import com.xy.job.admin.event.JobInstanceCreateEvent
import com.xy.job.admin.repository.JobInfoRepository
import com.xy.job.admin.repository.JobInstanceRepository
import com.xy.job.admin.repository.JobLockRepository
import com.xy.job.core.enums.JobStatus
import org.springframework.context.ApplicationEventPublisher
import org.springframework.scheduling.annotation.Scheduled
import org.springframework.scheduling.support.CronExpression
import org.springframework.stereotype.Component
import org.springframework.transaction.support.TransactionTemplate
import java.time.LocalDateTime
import java.util.concurrent.TimeUnit

@Component
class ScheduleJob(
    private var inetAddressHolder: InetAddressHolder,
    private val transactionTemplate: TransactionTemplate,
    private val jobLockRepository: JobLockRepository,
    private val jobInstanceRepository: JobInstanceRepository,
    private val jobInfoRepository: JobInfoRepository,
    private var eventPublisher: ApplicationEventPublisher,
) {

    @Scheduled(fixedRate = PRE_SCHEDULE_SECONDS, timeUnit = TimeUnit.SECONDS)
    fun schedule() {
        val preScheduledTime = LocalDateTime.now().plusSeconds(PRE_SCHEDULE_SECONDS)
        val lockNames = jobLockRepository.findLockNameByOwner(inetAddressHolder.address)
        val jobInfoGroup = jobInfoRepository.findAllScheduleable(lockNames, preScheduledTime).groupBy { it.lockName!! }
        jobInfoGroup.forEach { (lockName, jobInfos) ->
            // job_lock表提供细粒度行级锁，每个任务都与其中一把锁关联，获取到锁的节点可以执行锁对应的任务
            transactionTemplate.execute {
                jobLockRepository.lock(lockName)
                jobInfos.forEach { jobInfo ->
                    val existProcessingInstanceCount = jobInstanceRepository.countByJobIdAndJobStatus(
                        jobId = jobInfo.id!!,
                        jobStatus = JobStatus.PROCESSING,
                    )
                    val maxScheduleCount = jobInfo.maxConcurrentNum!! - existProcessingInstanceCount
                    if (maxScheduleCount > 0) {
                        scheduleJobInAdvanced(
                            jobInfo = jobInfo,
                            preScheduledTime = preScheduledTime,
                            maxScheduleCount = maxScheduleCount,
                        )
                    }
                }
            }
        }
    }

    private fun scheduleJobInAdvanced(
        jobInfo: JobInfo,
        preScheduledTime: LocalDateTime,
        maxScheduleCount: Int,
    ) {
        var nextTriggerTime: LocalDateTime? = null
        for (i in 0 until maxScheduleCount) {
            nextTriggerTime = scheduleJob(jobInfo, LocalDateTime.now())
            if (nextTriggerTime == null || nextTriggerTime <= preScheduledTime) {
                break
            }
        }
        if (jobInfo.nextTriggerTime != nextTriggerTime) {
            jobInfo.nextTriggerTime = nextTriggerTime
            jobInfoRepository.saveAndFlush(jobInfo)
        }
    }

    private fun scheduleJob(
        jobInfo: JobInfo,
        currentTime: LocalDateTime,
    ): LocalDateTime? {
        when {
            // 过去本应该被调度的任务由于调度系统宕机等原因没有被调度时，需要补偿该任务并基于当前时间重置下次调度时间
            jobInfo.nextTriggerTime!! <= currentTime -> {
                val nextTriggerTime = nextTriggerTime(
                    scheduleType = jobInfo.scheduleType!!,
                    scheduleConfig = jobInfo.scheduleConfig!!,
                    currentTime = LocalDateTime.now(),
                )
                val jobInstance = createJobInstance(jobInfo)
                eventPublisher.publishEvent(JobInstanceCreateEvent(jobInstance))
                return nextTriggerTime
            }

            // 预调度即将开始的任务
            jobInfo.nextTriggerTime!! <= currentTime.plusSeconds(PRE_SCHEDULE_SECONDS) -> {
                val nextTriggerTime = nextTriggerTime(
                    scheduleType = jobInfo.scheduleType!!,
                    scheduleConfig = jobInfo.scheduleConfig!!,
                    currentTime = jobInfo.nextTriggerTime!!,
                )
                val jobInstance = createJobInstance(jobInfo).apply {
                    this.scheduleTime = jobInfo.nextTriggerTime!!
                }
                eventPublisher.publishEvent(JobInstanceCreateEvent(jobInstance))
                return nextTriggerTime
            }

            // 更远的任务无需调度
            else -> return jobInfo.nextTriggerTime!!
        }
    }

    private fun createJobInstance(jobInfo: JobInfo): JobInstance {
        val jobInstance = jobInfo.createInstance()
        jobInstanceRepository.save(jobInstance)
        return jobInstance
    }

    fun nextTriggerTime(
        scheduleType: ScheduleType,
        scheduleConfig: String,
        currentTime: LocalDateTime = LocalDateTime.now(),
    ): LocalDateTime? {
        return when (scheduleType) {
            CRON -> CronExpression.parse(scheduleConfig).next(currentTime)!!
            FIX_RATE -> currentTime.plusSeconds(scheduleConfig.toLong())
            FIX_DELAY -> null
            ONE_TIME -> null
        }
    }
}